#include "ConfMgr.h"
#include <algorithm>
#include <fstream>
#include <iostream>
#include <set>
#include <string>
#include <vector>
#include "ErrorCode.h"
#include "InnerType.h"
#include "Util.h"
#include "XnccType.h"

namespace xncc {
namespace config {
    using nlohmann::json;

    int ConfMgr::loadCfgFile(const std::string& confFile)
    {
        std::string loadfile = confFile;
        if (loadfile.empty()) {
            loadfile = FILE_PATH_;
        }
        std::ifstream ifs(loadfile);
        if (!ifs.is_open()) {
            return -1;
        }
        json json;
        ifs >> json;
        platformInfo_ = json.get<PlatformInfo>();
        return 0;
    }

    int ConfMgr::loadCfgString(const std::string& cfgData)
    {
        if (cfgData.empty()) {
            return loadCfgFile(FILE_PATH_);
        }
        auto json     = nlohmann::json::parse(cfgData);
        platformInfo_ = json.get<PlatformInfo>();
        return 0;
    }

    ConfMgr::ConfMgr() : FILE_PATH_("../conf/conf.json") {}

    std::pair<int, std::string> ConfMgr::init()
    {
        auto ret = checkConf();
        if (ret.first != 0) {
            // check not pass
            return ret;
        }
        originRingQueues.clear();  // 没有用了
        genRealRingQueues();
        /// 对buffer进行分类
        classifyRingQueues();
        return {error::ERROR_CODE_SUCCESS, ""};
    }
    /*
        检查配置信息
    */
    std::pair<int, std::string> ConfMgr::checkConf()
    {
        auto ret = checkRingQueue();
        if (ret.first != error::ERROR_CODE_SUCCESS) {
            return ret;
        }
        ret = checkThreadId();
        if (ret.first != error::ERROR_CODE_SUCCESS) {
            return ret;
        }
        ret = checkAllBusiThread();
        if (ret.first != error::ERROR_CODE_SUCCESS) {
            return ret;
        }
        ret = checkAllIPCThread();
        if (ret.first != error::ERROR_CODE_SUCCESS) {
            return ret;
        }
        ret = checkQueueRace();
        if (ret.first != error::ERROR_CODE_SUCCESS) {
            return ret;
        }
        return {error::ERROR_CODE_SUCCESS, ""};
    }
    /*
        产生最小粒度的queue
    */
    void ConfMgr::classifyRingQueues()
    {
        for (auto& ele : realRingQueues) {
            auto ptr = ele.second;
            if (ptr->IsIpcThreadPub) {
                for (auto subId : ptr->SubTids) {
                    auto queue        = std::make_shared<ValidIPCRingQueueInfo>();
                    queue->Id         = ptr->CfgQueueInfo.Id;
                    queue->Size       = ptr->CfgQueueInfo.Size;
                    queue->SubTid     = subId;
                    queue->AutoExpand = ptr->CfgQueueInfo.AutoExpand;
                    ipcPubRingQueues[ptr->PubTids[0]].push_back(queue);
                }
            }
            else {
                for (auto subId : ptr->SubTids) {
                    auto queue        = std::make_shared<ValidBusiRingQueueInfo>();
                    queue->Id         = ptr->CfgQueueInfo.Id;
                    queue->Size       = ptr->CfgQueueInfo.Size;
                    queue->SubTid     = subId;
                    queue->MsgCodes   = ptr->MsgCodes;
                    queue->AutoExpand = ptr->CfgQueueInfo.AutoExpand;
                    busiPubRingQueues[ptr->PubTids[0]].push_back(queue);
                }
            }
        }
    }
    /*
        校验过了之后，产生真正的Queue配置信息
    */
    void ConfMgr::genRealRingQueues()
    {
        for (auto& queue : platformInfo_.RingQueues) {
            auto realQueue           = std::make_shared<RealRingQueueInfo>();
            realQueue->CfgQueueInfo  = queue;
            realRingQueues[queue.Id] = realQueue;
        }
        // 通过检验的Queue一定存在的
        for (auto& ele : platformInfo_.BusiThreads) {
            for (auto subId : ele.SubIds) {
                realRingQueues[subId]->SubTids.push_back(ele.Id);
            }
            // 处理业务线程的pub信息
            for (auto& pub : ele.PubIds) {
                realRingQueues[pub.QueueId]->PubTids.push_back(ele.Id);
                realRingQueues[pub.QueueId]->MsgCodes       = pub.MsgCodes;
                realRingQueues[pub.QueueId]->IsIpcThreadPub = false;
            }
        }
        for (auto& ele : platformInfo_.ServerThreads) {
            realRingQueues[ele.PubId]->PubTids.push_back(ele.Id);
            realRingQueues[ele.PubId]->IsIpcThreadPub = true;
            for (auto& subEle : ele.SubServer) {
                realRingQueues[subEle.PubId]->PubTids.push_back(subEle.Id);
                realRingQueues[subEle.PubId]->IsIpcThreadPub = true;
            }
        }

        for (auto& ele : platformInfo_.ClientThreads) {
            realRingQueues[ele.PubId]->PubTids.push_back(ele.Id);
            realRingQueues[ele.PubId]->IsIpcThreadPub = true;
        }

        // 一个id的订阅者和 生产者数量必须大于等于1，否则就是没有用的
        for (auto& ele : platformInfo_.RingQueues) {
            if (realRingQueues[ele.Id]->SubTids.empty() || realRingQueues[ele.Id]->PubTids.empty()) {
                std::cerr << "RingQueue = " << ele.Id
                          << ", subscriber and publisher are not all exist, will not gen real RingQueue" << '\n';
                realRingQueues.erase(ele.Id);
            }
        }
    }
    /*
        是否存在重复ID
    */
    template <typename T>
    bool ConfMgr::hasRepeated(const std::vector<T>& srcList)
    {
        std::set<T> tmpList(srcList.begin(), srcList.end());
        return tmpList.size() != srcList.size();
    }
    /*
        是否存在未定义的id
    */
    bool ConfMgr::hasUndefineId(const std::vector<int32_t>& idList)
    {
        return std::any_of(idList.begin(), idList.end(),
                           [this](int32_t iid) { return originRingQueues.find(iid) == originRingQueues.end(); });
    }
    /*
        是否存在自己订阅自己的现象，这样会导致死循环
    */
    bool ConfMgr::hasSelfSubPub(const std::vector<int32_t>& subIds, const std::vector<int32_t>& pubIds)
    {
        std::set<int> tmpSubIds(subIds.begin(), subIds.end());
        for (auto pid : pubIds) {
            if (tmpSubIds.find(pid) != tmpSubIds.end()) {
                return true;
            }
        }
        return false;
    }

    /*
        初始化操作暂不考虑效率问题
    */
    std::pair<int, std::string> ConfMgr::checkRingQueue()
    {
        // 第一不允许id重复
        // 第二不允许值小于等于0
        for (auto& queue : platformInfo_.RingQueues) {
            if (queue.Id <= 0) {
                return {error::QUEUE_ID_LESS_THAN_ZERO,
                        shared::concatToString("Id:", queue.Id, "  in Queues is not allowed to be less or equal zero")};
            }
            if (queue.Size < static_cast<int>(types::MIN_QUEUE_SIZE)) {
                return {error::QUEUE_ID_LESS_THAN_ZERO,
                        shared::concatToString("Id:", queue.Id, "  in Queues size:", queue.Size,
                                               " less than min size:", types::MIN_QUEUE_SIZE)};
            }
            originRingQueues[queue.Id] = queue;
        }
        if (originRingQueues.size() != platformInfo_.RingQueues.size()) {
            return {error::QUEUE_ID_REPEATED, "Id in Queues has repeated element"};
        }
        return {error::ERROR_CODE_SUCCESS, ""};
    }
    /*
        线程id，不允许重复
    */
    std::pair<int, std::string> ConfMgr::checkThreadId()
    {
        std::vector<int32_t> allThreadIds;

        for (auto& ele : platformInfo_.BusiThreads) {
            if (ele.Id <= 0) {
                return {error::BUSI_THREAD_ID_LESS_THAN_ZERO,
                        shared::concatToString("Id:", ele.Id, "  busi thread is not allowed to be less or equal zero")};
            }
            allThreadIds.push_back(ele.Id);
        }
        for (auto& ele : platformInfo_.ServerThreads) {
            if (ele.Id <= 0) {
                return {
                    error::SERVER_THREAD_ID_LESS_THAN_ZERO,
                    shared::concatToString("Id:", ele.Id, "  server thread is not allowed to be less or equal zero")};
            }
            allThreadIds.push_back(ele.Id);

            for (auto& subEle : ele.SubServer) {
                if (subEle.Id <= 0) {
                    return {error::SUB_SERVER_THREAD_ID_LESS_THAN_ZERO,
                            shared::concatToString("Id:", subEle.Id, "  sub server thread in server:", ele.Id,
                                                   " is not allowed to be less or equal zero")};
                }
                allThreadIds.push_back(subEle.Id);
            }
        }
        for (auto& ele : platformInfo_.ClientThreads) {
            if (ele.Id <= 0) {
                return {
                    error::CLIENT_THREAD_ID_LESS_THAN_ZERO,
                    shared::concatToString("Id:", ele.Id, "  client thread is not allowed to be less or equal zero")};
            }
            allThreadIds.push_back(ele.Id);
        }
        bool ret = hasRepeated(allThreadIds);
        if (ret) {
            return {error::THREAD_ID_REPEATED, "not allowded to config reapted thread id"};
        }
        return {error::ERROR_CODE_SUCCESS, ""};
    }
    /*
        检查是不是有多线程写同一个Queue，放在最后去校验，能走到这里，单个线程已经不存在重复id的现象
    */
    std::pair<int, std::string> ConfMgr::checkQueueRace()
    {
        std::unordered_map<int32_t, std::vector<int32_t>> allPubInIds;
        for (auto& ele : platformInfo_.BusiThreads) {
            for (auto& pub : ele.PubIds) {
                allPubInIds[pub.QueueId].push_back(ele.Id);
            }
        }
        for (auto& ele : platformInfo_.ServerThreads) {
            allPubInIds[ele.PubId].push_back(ele.Id);
            for (auto& subEle : ele.SubServer) {
                allPubInIds[subEle.PubId].push_back(subEle.Id);
            }
        }
        for (auto& ele : platformInfo_.ClientThreads) {
            allPubInIds[ele.PubId].push_back(ele.Id);
        }
        for (auto& ele : allPubInIds) {
            if (ele.second.size() > 1) {
                return {error::MULTI_THREADS_PUB_TO_SINGLE_QUEUE,
                        shared::concatToString("not allowed multi threads ",
                                               shared::CommonBusi::vectorToString(ele.second),
                                               " pub msg to one queue =", ele.first)};
            }
        }
        return {error::ERROR_CODE_SUCCESS, ""};
    }

    std::pair<int, std::string> ConfMgr::checkAllBusiThread()
    {
        for (auto& busi : platformInfo_.BusiThreads) {
            auto ret = checkBusiThread(busi);
            if (ret.first != 0) {
                return ret;
            }
        }
        return {error::ERROR_CODE_SUCCESS, ""};
    }

    std::pair<int, std::string> ConfMgr::checkBusiThread(const BusiThreadInfo& busi)
    {
        if (busi.IOMultiMode < 0 || busi.IOMultiMode > 3) {
            return {
                error::ERROR_IO_MULTI_MODE_IN_BUSI_THREAD,
                shared::concatToString("Id:", busi.Id, "  busi thread error IOMultiMode, you can choose 0, 1, 2, 3")};
        }
        // 检查订阅id是否存在,以及id是否重复
        bool ret = hasUndefineId(busi.SubIds);
        if (ret) {
            return {error::SUB_QUEUE_ID_IN_BUSI_THREAD_NOT_DEFINE,
                    shared::concatToString("Id:", busi.Id, "  busi thread has some sub id not define")};
        }
        ret = hasRepeated(busi.SubIds);
        if (ret) {
            return {error::SUB_QUEUE_ID_IN_BUSI_THREAD_HAS_REPEATED,
                    shared::concatToString("Id:", busi.Id, "  busi thread has some sub id reapted")};
        }
        std::vector<int32_t> tmpPubIds;
        tmpPubIds.reserve(busi.PubIds.size());
        for (const auto& pub : busi.PubIds) {
            tmpPubIds.push_back(pub.QueueId);
        }
        ret = hasUndefineId(tmpPubIds);
        if (ret) {
            return {error::PUB_QUEUE_ID_IN_BUSI_THREAD_NOT_DEFINE,
                    shared::concatToString("Id:", busi.Id, "  busi thread has some pub id not define")};
        }
        ret = hasRepeated(tmpPubIds);
        if (ret) {
            return {error::PUB_QUEUE_ID_IN_BUSI_THREAD_HAS_REPEATED,
                    shared::concatToString("Id:", busi.Id, "  busi thread has some pub id reapted")};
        }
        // 判断msgcode是否有重复
        for (const auto& pub : busi.PubIds) {
            for (auto code : pub.MsgCodes) {
                if (code <= static_cast<int>(shared::InnerMsgId::MAX_INNER_MSG_ID)) {  // 留给内部消息使用 0 1 2 3
                    return {error::PUB_MSG_CODE_IN_BUSI_THREAD_LESS_THAN_MIN_ID,
                            shared::concatToString("Id:", busi.Id, "  busi thread pub msg code is not allowed <=",
                                                   shared::InnerMsgId::MAX_INNER_MSG_ID)};
                }
            }
            ret = hasRepeated(pub.MsgCodes);
            if (ret) {
                return {error::PUB_MSG_CODE_IN_BUSI_THREAD_HAS_REPEATED,
                        shared::concatToString("Id:", busi.Id, "  busi thread  in pub id = ", pub.QueueId,
                                               " msg code has repeated ")};
            }
        }
        // 判断是否存在自己订阅自己的现象
        ret = hasSelfSubPub(busi.SubIds, tmpPubIds);
        if (ret) {
            return {error::SELF_PUB_SELF_SUB_IN_BUSI_THREAD,
                    shared::concatToString("Id:", busi.Id, "  busi thread  not allowded to sub and pub self ")};
        }

        if (busi.Name.empty()) {
            return {error::BUSI_THREAD_NAME_EMPTY,
                    shared::concatToString("Id:", busi.Id, "  busi thread name is empty ")};
        }
        return {error::ERROR_CODE_SUCCESS, ""};
    }
    std::pair<int, std::string> ConfMgr::checkAllIPCThread()
    {
        for (auto& ele : platformInfo_.ServerThreads) {
            auto ret = checkServerThread(ele);
            if (ret.first != 0) {
                return ret;
            }
        }
        for (auto& ele : platformInfo_.ClientThreads) {
            auto ret = checkClientThread(ele);
            if (ret.first != 0) {
                return ret;
            }
        }
        return {error::ERROR_CODE_SUCCESS, ""};
    }

    std::pair<int, std::string> ConfMgr::checkServerThread(const ServerThreadInfo& ioThread)
    {
        if (ioThread.IOMultiMode < 1 || ioThread.IOMultiMode > 3) {
            return {error::ERROR_IO_MULTI_MODE_IN_SERVER_THREAD,
                    shared::concatToString("Id:", ioThread.Id,
                                           "  server thread error IOMultiMode, you can choose 0, 1, 2, 3")};
        }
        std::vector<int32_t> tcpPubIds;
        tcpPubIds.push_back(ioThread.PubId);

        bool ret = hasUndefineId(tcpPubIds);
        if (ret) {
            return {error::PUB_QUEUE_ID_IN_SERVER_THREAD_NOT_DEFINE,
                    shared::concatToString("Id:", ioThread.Id, "  server thread has some sub id not define")};
        }
        if (ioThread.Name.empty()) {
            return {error::SERVER_THREAD_NAME_EMPTY,
                    shared::concatToString("Id:", ioThread.Id, "  server thread name empty")};
        }

        for (const auto& subEle : ioThread.SubServer) {
            auto ret = checkSubServerThread(subEle);
            if (ret.first != 0) {
                return ret;
            }
        }
        return {error::ERROR_CODE_SUCCESS, ""};
    }

    std::pair<int, std::string> ConfMgr::checkSubServerThread(const SubServerThreadInfo& ioThread)
    {
        if (ioThread.IOMultiMode < 1 || ioThread.IOMultiMode > 3) {
            return {error::ERROR_IO_MULTI_MODE_IN_SUB_SERVER_THREAD,
                    shared::concatToString("Id:", ioThread.Id,
                                           "  sub server thread error IOMultiMode, you can choose 0, 1, 2, 3")};
        }
        std::vector<int32_t> tcpPubIds;
        tcpPubIds.push_back(ioThread.PubId);

        bool ret = hasUndefineId(tcpPubIds);
        if (ret) {
            return {error::PUB_QUEUE_ID_IN_SUB_SERVER_THREAD_NOT_DEFINE,
                    shared::concatToString("Id:", ioThread.Id, " sub server thread has some sub id not define")};
        }
        if (ioThread.Name.empty()) {
            return {error::SUB_SERVER_THREAD_NAME_EMPTY,
                    shared::concatToString("Id:", ioThread.Id, " sub server thread name empty")};
        }
        return {error::ERROR_CODE_SUCCESS, ""};
    }

    std::pair<int, std::string> ConfMgr::checkClientThread(const ClientThreadInfo& ioThread)
    {
        if (ioThread.IOMultiMode < 1 || ioThread.IOMultiMode > 3) {
            return {error::ERROR_IO_MULTI_MODE_IN_CLIENT_THREAD,
                    shared::concatToString("Id:", ioThread.Id,
                                           "  client thread error IOMultiMode, you can choose 0, 1, 2, 3")};
        }
        std::vector<int32_t> tcpPubIds;
        tcpPubIds.push_back(ioThread.PubId);
        bool ret = hasUndefineId(tcpPubIds);
        if (ret) {
            return {error::PUB_QUEUE_ID_IN_CLIENT_THREAD_NOT_DEFINE,
                    shared::concatToString("Id:", ioThread.Id, " client thread has some sub id not define")};
        }
        if (ioThread.Name.empty()) {
            return {error::CLIENT_THREAD_NAME_EMPTY,
                    shared::concatToString("Id:", ioThread.Id, " client thread name empty")};
        }
        if (ioThread.ReConn && ioThread.ReConnIntervalMs < 0) {
            return {error::CLIENT_THREAD_RECONN_MS_LESS_THAN_ZERO,
                    shared::concatToString("Id:", ioThread.Id,
                                           " client thread ReConnIntervalMs config must >= 0 now is ",
                                           ioThread.ReConnIntervalMs)};
        }
        return {error::ERROR_CODE_SUCCESS, ""};
    }
}  // namespace config
}  // namespace xncc